package com.share.device.emqx.callback;

import com.alibaba.fastjson.JSON;
import com.share.common.core.constant.TokenConstants;
import com.share.device.emqx.factory.MessageHandlerFactory;
import com.share.device.emqx.handler.MessageHandler;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

/**
 * 回调消息处理类
 * @author zxl
 * @date 2025年03月28日 21:50
 */
@Slf4j
@Component
public class OnMessageCallback implements MqttCallback {

    @Resource
    private MessageHandlerFactory messageHandlerFactory;
    /**
     * 连接丢失
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        log.info("连接断开，可以做重连");
    }

    /**
     * 消息到达处理
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe（订阅）后得到的消息会执行到这里面
        log.info("接收消息主题：{}", topic);
        log.info("接收消息Qos：{}", message.getQos());
        log.info("接收消息内容：{}", new String(message.getPayload()));
        try {
            // 根据主题选择不同的处理逻辑
            MessageHandler messageHandler = messageHandlerFactory.getMessageHandler(topic);
            if (null != messageHandler) {
                String content = new String(message.getPayload());
                messageHandler.handleMessage(JSON.parseObject(content));
            }
        } catch (Exception e) {
                e.printStackTrace();
                log.error("mqtt消息异常：{}", new String(message.getPayload()));
        }
    }

    /**
     * 消息发送完成处理
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("deliveryComplete--------{}", token.isComplete());
    }
}
